package com.hery.functions;

import com.hery.pojo.KafkaAlertLog;
import com.hery.pojo.Rule;
import com.hery.utils.BeanToMap;
import com.hery.utils.Config;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;

import java.io.IOException;
import java.time.LocalDateTime;
import java.util.Map;
import java.util.Properties;


public class SinkFunctions {

    private static final String clientEventInsertSql = "INSERT INTO `flink`.`flink-alert-data` " +
            "(`app_package`, `collected_time`, `behavior_id`, `qisiui_version`, `uid`, `region_code`, `os_version`, `event_name`, `vin`, `hardware_version`, `carseries`, `pdsn`, `display_id`, `rule_name`, `rule_id`, `rule_exp`) " +
            "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";

    private static final String cloudInsertSql = "insert into `flink`.`flink-cloud-alert-data`(`microservice`, `reqPath`, `clicnetIP`, `resultCode`, `createDate`, `ctime`, `rule_name`, `rule_id`, `rule_exp`) values (?, ?, ?, ?, ?, ?, ?, ?, ?) ";

    public static FlinkKafkaProducer<String> getFlinkKafkaProducer() throws IOException {

        Properties prop = new Properties();
        prop.load(SinkFunctions.class.getClassLoader().getResourceAsStream(Config.KAFKA_PRODUCER_CONFIG_FILE));

        return new FlinkKafkaProducer<>(
                Config.KAFKA_PRODUCER_TOPIC,
                new SimpleStringSchema(),
                prop
        );
    }

    public static SinkFunction<KafkaAlertLog> getMysqlAlertSinkFunction(Configuration config) throws IOException {

        Config.initApplicationConfig(config);

        JdbcConnectionOptions options = new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                .withUrl(Config.MYSQL_URL)
                .withDriverName(Config.MYSQL_DRIVER_CLASS)
                .withUsername(Config.MYSQL_USER)
                .withPassword(Config.MYSQL_PSWD)
                .build();

        JdbcExecutionOptions executionOptions = JdbcExecutionOptions.builder()
                .withBatchSize(100)
                .withBatchIntervalMs(200)
                .withMaxRetries(5)
                .build();

        switch (Config.APP_TYPE){

            case CLIENT_EVENT:  // 车机端埋点
                return getMysqlEventSink(options, executionOptions);

            case SENTRY: // Sentry 异常上报日志
                return getMysqlSentrySink(options, executionOptions);

            case CLOUD_LOG:   // 云端微服务日志
                return getMysqlCloudLogSink(options, executionOptions);

            case CLOUD_EVENT: // 云端cp埋点日志
                return getMysqlCloudEventSink(options, executionOptions);

            default: throw new RuntimeException("unknown app.type, route can not null ");

        }

    }

    /**
     * Sentry 异常上报 MySQL Sink
     * @return
     */
    private static SinkFunction<KafkaAlertLog> getMysqlSentrySink(JdbcConnectionOptions options, JdbcExecutionOptions executionOptions) {

        JdbcStatementBuilder<KafkaAlertLog> statementBuilder = (statement, log) -> {
            Rule rule = log.getRule();
            Map<String, Object> map = BeanToMap.jsonFlatten(log.getJson());

            statement.setString(1, map.getOrDefault("microservice", "").toString());
            statement.setString(2, map.getOrDefault("reqPath", "").toString());
            statement.setString(3, map.getOrDefault("clicnetIP", "").toString());
            statement.setString(4, map.getOrDefault("statusCode", "").toString());
            statement.setString(5, map.getOrDefault("logTime", "").toString());
            statement.setString(6, map.getOrDefault("ctime", LocalDateTime.now()).toString());
            statement.setString(7, rule.getName());
            statement.setString(8, rule.getId() + "");
            statement.setString(9, rule.getExp());
        };

        return JdbcSink.sink( cloudInsertSql, statementBuilder,  executionOptions, options  );
    }


    /**
     * 车机端埋点日志 Sink
     * 参考：https://github.com/ClickHouse/clickhouse-jdbc
     */
    private static SinkFunction<KafkaAlertLog> getMysqlEventSink(JdbcConnectionOptions options, JdbcExecutionOptions executionOptions) {

        JdbcStatementBuilder<KafkaAlertLog> statementBuilder = (statement, log) -> {
            KafkaAlertLog alertLog = (KafkaAlertLog) log;
            Rule rule = alertLog.getRule();
            Map<String, Object> map = BeanToMap.jsonFlatten(log.getJson());

            statement.setString(1, map.getOrDefault("common_data.appPackage", "").toString());
            statement.setLong(2, Long.parseLong(map.getOrDefault("common_data.collectedTime", "0").toString()));
            statement.setString(3, map.getOrDefault("common_data.behaviorId", "").toString());
            statement.setString(4, map.getOrDefault("common_data.qisiuiVersion", "").toString());
            statement.setString(5, map.getOrDefault("common_data.uid", "").toString());
            statement.setString(6, map.getOrDefault("common_data.regionCode", "").toString());
            statement.setString(7, map.getOrDefault("common_data.appVersion", "").toString());
            statement.setString(8, map.getOrDefault("common_data.eventName", "").toString());
            statement.setString(9, map.getOrDefault("common_data.vin", "").toString());
            statement.setString(10, map.getOrDefault("common_data.hardwareVersion", "").toString());
            statement.setString(11, map.getOrDefault("common_data.carseries", "").toString());
            statement.setString(12, map.getOrDefault("common_data.pdsn", "").toString());
            statement.setString(13, map.getOrDefault("common_data.displayId", "").toString());
            statement.setString(14, rule.getName());
            statement.setString(15, rule.getId() + "");
            statement.setString(16, rule.getExp());
        };
        System.out.println(clientEventInsertSql);
        return JdbcSink.sink( clientEventInsertSql,  statementBuilder, executionOptions,  options );
    }


    /**
     * @return 云端微服务日志数据表
     */
    private static SinkFunction<KafkaAlertLog> getMysqlCloudLogSink(JdbcConnectionOptions options, JdbcExecutionOptions executionOptions) {

        JdbcStatementBuilder<KafkaAlertLog> statementBuilder = (statement, log) -> {
            KafkaAlertLog alertLog = (KafkaAlertLog) log;
            Rule rule = alertLog.getRule();
            Map<String, Object> map = BeanToMap.jsonFlatten(log.getJson());

            statement.setString(1, map.getOrDefault("appName", "").toString());
            statement.setString(2, map.getOrDefault("message.reqPath", "").toString());
            statement.setString(3, map.getOrDefault("message.clicnetIP", "").toString());
            statement.setString(4, map.getOrDefault("message.resultCode", "").toString());
            statement.setString(5, map.getOrDefault("logTime", "").toString());
            statement.setString(6, map.getOrDefault("ctime", LocalDateTime.now()).toString());
            statement.setString(7, rule.getName());
            statement.setString(8, rule.getId() + "");
            statement.setString(9, rule.getExp());
        };

        return JdbcSink.sink( cloudInsertSql, statementBuilder,  executionOptions, options  );
    }

    /**
     * CP 埋点数据
     * @return
     */
    private static SinkFunction<KafkaAlertLog> getMysqlCloudEventSink(JdbcConnectionOptions options, JdbcExecutionOptions executionOptions) {

        JdbcStatementBuilder<KafkaAlertLog> statementBuilder = (statement, log) -> {
            Rule rule = log.getRule();
            Map<String, Object> map = BeanToMap.jsonFlatten(log.getJson());

            statement.setString(1, map.getOrDefault("microservice", "").toString());
            statement.setString(2, map.getOrDefault("cp_provider_access_url", "0").toString());
            statement.setString(3, map.getOrDefault("clicnetIP", "").toString());
            statement.setString(4, map.getOrDefault("statusCode", "").toString());
            statement.setString(5, map.getOrDefault("createDate", "").toString());
            statement.setString(6, map.getOrDefault("ctime", LocalDateTime.now()).toString());
            statement.setString(7, rule.getName());
            statement.setString(8, rule.getId() + "");
            statement.setString(9, rule.getExp());
        };


        return JdbcSink.sink( cloudInsertSql, statementBuilder,  executionOptions, options  );
    }

}
